package com.linsy.product.redis.mq.service;

import cn.hutool.core.collection.CollUtil;
import com.linsy.product.redis.mq.config.KernelMqDataConfig;
import com.linsy.product.redis.mq.entity.MqDataVO;
import com.linsy.product.redis.mq.util.RedisUtil;

import java.lang.reflect.InvocationTargetException;
import java.util.List;
import java.util.concurrent.ConcurrentMap;

/**
 * 消费者装饰器
 */
public abstract class ConsumersMsgDecorator<T> implements RedisMqConsumers<T> {

    /**
     * 消费主题
     */
    private final String topic;
    private final RedisUtil redisUtil;

    /**
     * @param topic 主题必填
     */
    public ConsumersMsgDecorator(RedisUtil redisUtil, String topic) {
        this.redisUtil = redisUtil;
        this.topic = topic;
    }

    /**
     * <b>必须实现</b>
     * 业务消费方法
     */
    @Override
    public void consumersMethod(T t) {

    }

    /**
     * <b>不要实现</b>
     * 内部消费方法
     */
    @Override
    public void initConsumers() throws Exception {
        ConcurrentMap<String, Thread> topicRunState = KernelMqDataConfig.TOPIC_RUN_STATE;
        // 保证只有一条线程在跑这个主题
        while (!topicRunState.containsKey(topic) || Thread.currentThread().equals(topicRunState.get(topic))) {
            // 标记有线程正在运行此主题
            topicRunState.put(topic, Thread.currentThread());

            List<MqDataVO> value = redisUtil.pullTask(topic, 10L, 1L);
            if (CollUtil.isEmpty(value)) {
                topicRunState.remove(topic);
                return;
            }

            for (MqDataVO mqDataVO : value) {
                try {
                    beforeConsumers(mqDataVO);
                    consumersMethod(this.convert(mqDataVO.getMqData()));
                    afterConsumers(mqDataVO);
                } catch (Exception e) {
                    InvocationTargetException targetEx = (InvocationTargetException)e;
                    Throwable trowEx = targetEx .getTargetException();
                    consumersError(trowEx);
                }
            }
        }
    }

}
